﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

#region

using System;
using System.Collections.Generic;
using System.IO;
using System.Runtime.CompilerServices;
using AzureMRCore.DataTypes;
using AzureMRCore.Primitives;

#endregion

namespace AzureMRCore.InputFormat
{
    public class CachedDoubleVectorInputFormat : IInputFormat<IntKey, DoubleVectorValue>, ICachedInputFormat
    {
        private int _counter;
        private DoubleVectorValue _currentVector;
        private Boolean _isDeffering = true, _isKeyed;
        private List<KeyValuePair<IntKey, DoubleVectorValue>> _lines;
        private int _numVectors;
        private BinaryReader _reader;
        private int _row, _rowOffset;
        private long _size;
        private int _vectorLength;

        #region ICachedInputFormat Members

        public long GetSize()
        {
            return _size;
        }

        public ICachedInputFormat GetCachedCopy()
        {
            ReadAll();
            CachedDoubleVectorInputFormat newCachedFormat = new CachedDoubleVectorInputFormat();
            newCachedFormat.Init(_lines, _size, _vectorLength, _numVectors, _row, _rowOffset);
            return newCachedFormat;
        }

        #endregion

        #region IInputFormat<IntKey,DoubleVectorValue> Members

        public void Init(string inputFile)
        {
            _reader = new BinaryReader(File.Open(inputFile, FileMode.Open));
            FileInfo file = new FileInfo(inputFile);
            _size = file.Length;
            _numVectors = _reader.ReadInt32();
            _vectorLength = _reader.ReadInt32();
            _row = _reader.ReadInt32();
            _rowOffset = _reader.ReadInt32();
            _isKeyed = _reader.ReadBoolean();
            _lines = new List<KeyValuePair<IntKey, DoubleVectorValue>>();
        }

        public List<KeyValuePair<IntKey, DoubleVectorValue>> GetAllPairs()
        {
            return ReadAll();
        }

        public IntKey GetCurrentKey()
        {
            IntKey intKey = new IntKey {Key = _counter};
            return intKey;
        }

        public DoubleVectorValue GetCurrentValue()
        {
            return _currentVector;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public bool Next()
        {
            if (_counter >= _numVectors)
            {
                _isDeffering = false;
                return false;
            }
            if (_isDeffering)
            {
                double[] vector = new double[_vectorLength];
                int key = _reader.ReadInt32();
                for (int i = 0; i < _vectorLength; i++)
                {
                    vector[i] = _reader.ReadDouble();
                }
                _currentVector = DoubleVectorValue.GetInstance(vector);
                _lines.Add(new KeyValuePair<IntKey, DoubleVectorValue>(IntKey.GetInstance(key), _currentVector));
                _counter++;
                return true;
            }
            _currentVector = _lines[_counter].Value;
            _counter++;
            return true;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public bool HasNext()
        {
            return _numVectors > _counter;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public void Close()
        {
            if (_isDeffering)
            {
                ReadAll();
                _reader.Close();
                _isDeffering = false;
            }
//            _reader.Close();
        }

        #endregion

        public void Init(IDictionary<int, BCastDataProduct> inputRecords)
        {
            throw new NotImplementedException();
        }

        private void Init(List<KeyValuePair<IntKey, DoubleVectorValue>> lines, long size, int vectorLength,
                          int numVectors, int row, int rowOffset)
        {
            _lines = lines;
            _size = size;
            _isDeffering = false;
            _vectorLength = vectorLength;
            _numVectors = numVectors;
            _row = row;
            _rowOffset = rowOffset;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        private List<KeyValuePair<IntKey, DoubleVectorValue>> ReadAll()
        {
            if (!_isDeffering) return _lines;
            while (_counter < _numVectors)
            {
                int key = _reader.ReadInt32();
                double[] vector = new double[_vectorLength];
                for (var i = 0; i < _vectorLength; i++)
                {
                    vector[i] = _reader.ReadDouble();
                }
                _lines.Add(new KeyValuePair<IntKey, DoubleVectorValue>(IntKey.GetInstance(key),
                                                                       DoubleVectorValue.GetInstance(vector)));
                _counter++;
            }
            _isDeffering = false;
            return _lines;
        }
    }
}